fix(experimental): implement requests_done method to signal end of requests in async streams. Gracefully close streams. #1700
Conversation
…quests in async streams. Gracefully close streams.
product-auto-label
bot
added
size: m
Summary of ChangesHello @chandra-siri, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This pull request enhances the robustness of experimental asynchronous read and write object streams by implementing a mechanism to explicitly signal the end of client requests. This change ensures that the bidi-gRPC connection is closed gracefully, preventing potential issues where the server might be left waiting for further client input. The addition of dedicated unit tests validates this new behavior, contributing to the overall stability of the async stream handling. Highlights
🧠 New Feature in Public Preview: You can now enable Memory to help Gemini Code Assist learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. Click here to enable Memory in your admin console. Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up here. You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension. Footnotes
|
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces a requests_done method to gracefully close asynchronous streams for reading and writing objects. The implementation correctly signals the end of requests to the server before closing the connection. My review focuses on improving code quality and test coverage. I've suggested adding guard clauses to the new methods for robustness and refactoring duplicated code into a base class to improve maintainability. I've also recommended adding tests for the failure scenarios of the new methods.
| async def requests_done(self): | ||
| """Signals that all requests have been sent.""" | ||
|
|
||
| await self.socket_like_rpc.send(None) | ||
| await self.socket_like_rpc.recv() |
There was a problem hiding this comment.
This new method has a couple of areas for improvement:
- Missing guard clause: For consistency with other methods like
sendandrecv, a check should be added to ensure the stream is open before proceeding. This prevents errors if the method is called on a closed stream. - Code duplication: This method is identical to the one in
_AsyncWriteObjectStream. To improve maintainability, consider moving this shared logic to the_AsyncAbstractObjectStreambase class. This would likely involve moving_is_stream_openandsocket_like_rpcto the base class as well.
Here is a suggested implementation that includes the guard clause:
async def requests_done(self):
"""Signals that all requests have been sent."""
if not self._is_stream_open:
raise ValueError("Stream is not open")
await self.socket_like_rpc.send(None)
await self.socket_like_rpc.recv()| async def requests_done(self): | ||
| """Signals that all requests have been sent.""" | ||
|
|
||
| await self.socket_like_rpc.send(None) | ||
| await self.socket_like_rpc.recv() |
There was a problem hiding this comment.
Similar to my comment on async_read_object_stream.py, this method should include a check to ensure the stream is open.
Additionally, this logic is duplicated across both stream classes. Refactoring this into the _AsyncAbstractObjectStream base class would be a good improvement for maintainability.
async def requests_done(self):
"""Signals that all requests have been sent."""
if not self._is_stream_open:
raise ValueError("Stream is not open")
await self.socket_like_rpc.send(None)
await self.socket_like_rpc.recv()| "google.cloud.storage._experimental.asyncio.async_grpc_client.AsyncGrpcClient.grpc_client" | ||
| ) | ||
| @pytest.mark.asyncio | ||
| async def test_requests_done(mock_client, mock_cls_async_bidi_rpc): |
There was a problem hiding this comment.
This is a good test for the happy path. With the recommended addition of the _is_stream_open check in requests_done, it would be beneficial to also add a test case to verify that a ValueError is raised when requests_done is called on a closed stream, similar to test_close_without_open_should_raise_error.
| @mock.patch( | ||
| "google.cloud.storage._experimental.asyncio.async_write_object_stream.AsyncBidiRpc" | ||
| ) | ||
| async def test_requests_done(mock_cls_async_bidi_rpc, mock_client): |
There was a problem hiding this comment.
Similar to the read stream test, this is a good happy path test. It would be valuable to add another test to ensure requests_done raises a ValueError when the stream is not open, to cover the failure case. This would be analogous to test_close_without_open_should_raise_error.
2 participants
fix(experimental): implement requests_done method to signal end of requests in async streams. Gracefully close streams.